package com.tiantang.kafka.stream;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class LogProcessor implements Processor<byte[],byte[]>{
	
	private static Logger logger=LoggerFactory.getLogger(LogProcessor.class);
	
	private ProcessorContext context=null;

	public void init(ProcessorContext context) {
		this.context=context;
	}

	/**
	 * 
	 * @Title: process  
	 * @description:功能描述 流式处理的处理方法
	 * @param key
	 * @param value   
	 * @author:liujinkun
	 * @date 2018年5月28日
	 */
	public void process(byte[] key, byte[] value) {
		logger.info("kafka流式处理消息");
		String valueStr=new String(value);
		// 将 ** 替换为 ^^^
		String value2=valueStr.replaceAll("\\*\\*", "^^^");
		context.forward(key, value2.getBytes());
	}

	public void punctuate(long timestamp) {
		
	}

	public void close() {
		
	}

}
